package com.atguigu.app.dimdwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.func.DimDwdTableProcessFunction;
import com.atguigu.app.func.DimSinkFunction;
import com.atguigu.bean.TableProcess;
import com.atguigu.utils.MyKafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * ClassName: BaseDBApp
 * Package: com.atguigu.app.dimdwd
 * Description:
 *
 * @Author 尚硅谷-李增亮
 * @Create 2023/5/16 9:01
 * @Version 1.0
 */
//数据流:web/app -> MySQL -> Maxwell -> Kafka(ODS) -> FlinkApp -> Kafka(DWD)
//数据流:web/app -> Mysql -> Maxwell -> Kafka(ODS) -> FlinkApp -> Phoenix(DIM)
//程 序:Mock -> Mysql -> Maxwell -> Kafka(ZK) -> BaseDBApp(FlinkCDC) -> Phoenix(HBase ZK/HDFS)
//程 序:mock -> Mysql -> Maxwell -> Kafka(ZK) -> BaseDBApp(FlinkCDC) -> Kafka(ZK)
public class BaseDBApp {
    public static void main(String[] args) throws Exception {
//  1. 获取运行环境，设置并行度
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //  如果需要保证精确一次性，需要开启检查点
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //  TODO 2.读取Kafka ODS层 topic_db 主题数据
        DataStreamSource<String> kafkaDS = env.fromSource(MyKafkaUtil.getKafkaSource("topic_db", "base_db_app_221109"), WatermarkStrategy.noWatermarks(), "kafka-source");

        kafkaDS.print("kafkaDS>>>>>");

        //  TODO 3. 过滤并转化为JSON对象，如果需要侧输出流只能使用process方法
        SingleOutputStreamOperator<JSONObject> josnObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out)  {
                if (value != null) {
                    try {
                        //  解析value格式
                        JSONObject jsonObject = JSON.parseObject(value);
                        out.collect(jsonObject);
                    } catch (JSONException e) {
                        //  控制异常力度
                        //  防止因为存在脏数据导致程序终止
                        System.out.println("脏数据:" + value);
                    }
                }
            }
        });

        //  TODO 4.使用FlinkCDC读取配置表
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("000000")
                .databaseList("edu_config")
                .tableList("edu_config.table_process")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();
        DataStreamSource<String> mysqlDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-source");
//        mysqlDS.print("--->");

        //  TODO 5. 将配置流转化为广播流
        MapStateDescriptor<String, TableProcess> mapStateDescriptor = new MapStateDescriptor<>("dwd-map-state", String.class, TableProcess.class);

        BroadcastStream<String> broadcastDS = mysqlDS.broadcast(mapStateDescriptor);

        //  TODO 6. 连接主流与广播流
        BroadcastConnectedStream<JSONObject, String> connectedStream = josnObjDS.connect(broadcastDS);

        //TODO 7.处理连接流
        OutputTag<JSONObject> outputTag = new OutputTag<JSONObject>("hbase") {
        };
        SingleOutputStreamOperator<JSONObject> processDS = connectedStream.process(new DimDwdTableProcessFunction(mapStateDescriptor, outputTag));



        //TODO 8.将主流数据写出到Kafka,同时提取侧输出流数据写出到Phoenix
        processDS.print("processDS>>>>>>>>");
        processDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject element, @Nullable Long timestamp) {
                return new ProducerRecord<>(element.getString("sink_table"),
                        element.getString("data").getBytes());
            }
        }));
        DataStream<JSONObject> hbaseDS = processDS.getSideOutput(outputTag);
        hbaseDS.print("hbase>>>>>>>>>");
        hbaseDS.addSink(new DimSinkFunction());

        //TODO 9.启动任务
        env.execute("BaseDBApp");
    }
}
